package org.ws.task.zookeeper.node;

import org.apache.commons.lang.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.curator.framework.recipes.leader.LeaderLatchListener;
import org.apache.curator.framework.recipes.leader.Participant;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.ws.task.Bootable;
import org.ws.task.common.IPUtils;
import org.ws.task.model.NodeMetadata;
import org.ws.task.model.NodeRole;
import org.ws.task.model.NodeStatus;
import org.ws.task.repository.NodeMetadataRepository;
import org.ws.task.zookeeper.NodePaths;
import org.ws.task.zookeeper.listener.NodeListenerManager;

import java.io.IOException;
import java.util.Date;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

public class LeaderNode extends AbstractNode implements Bootable {

    private static Logger log = LoggerFactory.getLogger(LeaderNode.class.getName());

    private final LeaderLatch leaderLatch;
    private final CuratorCache leaderDataCache;
    private final CountDownLatch countDownLatch;
    private final DefaultLeaderLatchListener defaultLeaderLatchListener;
    private final AtomicReference<NodeMetadata> currentNodeMetadata;
    private final AtomicBoolean isLeader;

    public LeaderNode(String id, String namespace, CuratorFramework curatorClient, NodeListenerManager nodeListenerManager) {
        super(id, namespace, curatorClient, nodeListenerManager);
        this.leaderLatch = new LeaderLatch(this.curatorClient, NodePaths.getLeaderPath(namespace),this.id);
        this.countDownLatch = new CountDownLatch(1);
        this.leaderDataCache = CuratorCache.build(this.curatorClient, NodePaths.getLeaderDataPath(namespace));

        this.currentNodeMetadata = new AtomicReference<>();
        this.isLeader = new AtomicBoolean(false);
        this.defaultLeaderLatchListener = new DefaultLeaderLatchListener();

        this.leaderDataCache.listenable().addListener(buildLeaderDataListener());
        this.leaderLatch.addListener(this.defaultLeaderLatchListener);
    }

    @Override
    public void start() {
        this.leaderDataCache.start();
        try {
            this.leaderLatch.start();
            // 阻塞等待完成竞选
            while (this.countDownLatch.getCount()>0){
                Participant leader = this.leaderLatch.getLeader();
                // 竞选完成
                if(leader != null && StringUtils.isNotBlank(leader.getId())){
                    if(!StringUtils.equals(leader.getId(),this.id)){
                        this.defaultLeaderLatchListener.notLeader();
                    }
                    else {
                        this.defaultLeaderLatchListener.isLeader();
                    }
                }
                else {
                    Thread.sleep(200);
                }
            }
        } catch (Exception e) {
            log.error(e.getMessage(),e);
        }
        log.info("Leader Node startup.[{}]",this.isLeader.get());
    }

    @Override
    public void shutdown() {
        this.leaderDataCache.close();
        try {
            this.leaderLatch.close();
        } catch (IOException e) {
            log.error(e.getMessage(),e);
        }
    }

    /**
     * 本节点为Leader时更新节点数据
     * server服务启动后更新节点状态
     */
    public void updateLeader(int port, NodeStatus status){
        NodeMetadata nodeMetadata = this.currentNodeMetadata.get();
        if(nodeMetadata == null){
            nodeMetadata = NodeMetadata.builder()
                    .id(this.id)
                    .host(IPUtils.getIP())
                    .port(-1)
                    .role(NodeRole.LEADER)
                    .status(NodeStatus.READY)
                    .updateTime(new Date())
                    .build();
        }
        nodeMetadata.setStatus(status);
        nodeMetadata.setPort(port);
        nodeMetadata.setUpdateTime(new Date());
        this.currentNodeMetadata.set(nodeMetadata);
        String path = NodePaths.getLeaderDataPath(namespace);
        try {
            this.updateNodeData(path,nodeMetadata, CreateMode.PERSISTENT);
        } catch (Exception e) {
            log.error(e.getMessage(),e);
            // TODO: 更新失败情况
        }
    }

    /**
     * 获取Leader
     * @return
     */
    public NodeMetadata getLeader(){
        if(this.currentNodeMetadata.get() == null){
            try {
                byte[] data = this.curatorClient.getData().forPath(NodePaths.getLeaderDataPath(namespace));
                NodeMetadata nodeMetadata = NodeDataCodec.decode(data, NodeMetadata.class);
                this.nodeListenerManager.notLeader(nodeMetadata);
            }
            catch (Exception e){
                log.error(e.getMessage(),e);
            }
        }
        return this.currentNodeMetadata.get();
    }

    /**
     * 当前实例是否是Leader
     * @return
     */
    public boolean isLeader(){
        return this.isLeader.get();
    }

    /**
     * 监听LeaderData获取nodeMetadata
     * @return
     */
    private CuratorCacheListener buildLeaderDataListener(){
        return CuratorCacheListener.builder()
                .forCreatesAndChanges((oldNode, child) -> {
                    if(child != null){
                        byte[] data = child.getData();
                        NodeMetadata nodeMetadata = NodeDataCodec.decode(data, NodeMetadata.class);
                        NodeMetadata leader = NodeMetadataRepository.getInstance().getLeader();
                        if(nodeMetadata != null){
                            log.debug("changeLeaderNode:{},{}",leader==null?"null":leader.getId(),nodeMetadata.getId());
                            LeaderNode.this.nodeListenerManager.changeLeaderNode(nodeMetadata);
                            LeaderNode.this.currentNodeMetadata.set(nodeMetadata);
                        }
                    }
                })
                .forDeletes(child -> {
                    if(child != null) {
                        byte[] delChildDataData = child.getData();
                        NodeMetadata nodeMetadata = NodeDataCodec.decode(delChildDataData, NodeMetadata.class);
                        if (nodeMetadata != null) {
                            // 移除当前leader
                            NodeMetadata leader = NodeMetadataRepository.getInstance().getLeader();
                            if (leader != null && StringUtils.equals(leader.getId(), nodeMetadata.getId())) {
                                log.debug("removeLeaderNode:{}=={}", leader.getId(), nodeMetadata.getId());
                                LeaderNode.this.nodeListenerManager.removeLeaderNode(nodeMetadata);
                            }
                        }
                    }
                })
                .build();
    }

    class DefaultLeaderLatchListener implements LeaderLatchListener {
        @Override
        public void isLeader() {
            try {
                log.info("is leader:{}", LeaderNode.this.id);
                LeaderNode.this.isLeader.set(true);
                try {
                    LeaderNode.this.currentNodeMetadata.set(null);
                    LeaderNode.this.updateLeader(-1, NodeStatus.READY);
                } catch (Exception e) {
                    log.error(e.getMessage(),e);
                }
                finally {
                    LeaderNode.this.nodeListenerManager.becomeLeader(LeaderNode.this.currentNodeMetadata.get());
                }
            }
            finally {
                LeaderNode.this.countDownLatch.countDown();
            }
        }

        @Override
        public void notLeader() {
            try {
                log.info("not leader:{}", LeaderNode.this.id);
                LeaderNode.this.isLeader.set(false);
                byte[] data = LeaderNode.this.curatorClient.getData().forPath(NodePaths.getLeaderDataPath(namespace));
                NodeMetadata nodeMetadata = NodeDataCodec.decode(data, NodeMetadata.class);
                LeaderNode.this.currentNodeMetadata.set(nodeMetadata);
                LeaderNode.this.nodeListenerManager.notLeader(nodeMetadata);
            }
            catch (Exception e) {
                log.error(e.getMessage(),e);
                // TODO: Event
            }
            finally {
                LeaderNode.this.countDownLatch.countDown();
            }
        }
    }
}
